package cn.iocoder.yudao.module.mqtt.framework.mqtt;

import cn.iocoder.yudao.module.mqtt.dal.dataobject.mqtt.MqttConfigDO;
import cn.iocoder.yudao.module.mqtt.framework.config.MqttReconnectProperties;
import cn.iocoder.yudao.module.mqtt.framework.handler.InstanceSpecificMessageHandler;
import cn.iocoder.yudao.module.mqtt.framework.handler.MultiMqttMessageHandler;
import cn.iocoder.yudao.module.mqtt.service.subscribewater.SubscribeWaterService;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttClient;

/**
 * MQTT实例管理器
 * 用于管理多个MQTT连接实例
 *
 * @author 芋道源码
 */
@Slf4j
@Component
public class MqttInstanceManager {

    @Autowired
    private cn.iocoder.yudao.module.mqtt.service.mqtt.MqttMessageService mqttMessageService;

    @Autowired
    private SubscribeWaterService subscribeWaterService;

    @Autowired
    private MultiMqttMessageHandler multiMqttMessageHandler;

    @Autowired
    private cn.iocoder.yudao.module.mqtt.service.mqtt.MqttConfigService mqttConfigService;

    @Autowired
    private MqttConnectionListener mqttConnectionListener;

    @Autowired
    private MqttReconnectProperties reconnectProperties;

    /**
     * MQTT客户端工厂缓存
     */
    private final Map<Long, MqttPahoClientFactory> clientFactories = new ConcurrentHashMap<>();

    /**
     * MQTT入站适配器缓存
     */
    private final Map<Long, MqttPahoMessageDrivenChannelAdapter> inboundAdapters = new ConcurrentHashMap<>();

    /**
     * MQTT出站处理器缓存
     */
    private final Map<Long, MqttPahoMessageHandler> outboundHandlers = new ConcurrentHashMap<>();

    /**
     * 入站消息通道缓存
     */
    private final Map<Long, MessageChannel> inboundChannels = new ConcurrentHashMap<>();

    /**
     * 出站消息通道缓存
     */
    private final Map<Long, MessageChannel> outboundChannels = new ConcurrentHashMap<>();

    /**
     * 消息处理器缓存
     */
    private final Map<Long, InstanceSpecificMessageHandler> messageHandlers = new ConcurrentHashMap<>();

    /**
     * 重连状态缓存
     */
    private final Map<Long, AtomicBoolean> reconnectingFlags = new ConcurrentHashMap<>();

    /**
     * 定时任务执行器
     */
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);

    /**
     * 重连配置（使用配置属性中的值）
     */

    /**
     * 初始化默认MQTT实例
     */
    @PostConstruct
    public void init() {
        log.info("初始化MQTT实例管理器");
        // 不在这里初始化实例，由MultiMqttConfiguration负责
    }

    /**
     * 应用关闭时清理资源
     */
    @PreDestroy
    public void destroy() {
        log.info("开始清理MQTT实例管理器资源...");
        try {
            stopAllInstances();
            // 关闭定时任务执行器
            if (!scheduler.isShutdown()) {
                scheduler.shutdown();
                if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
                    scheduler.shutdownNow();
                }
            }
        } catch (Exception e) {
            log.error("清理MQTT实例管理器资源时发生异常", e);
        }
        log.info("MQTT实例管理器资源清理完成");
    }

    // 日志回调工厂
    public MqttPahoClientFactory createClientFactory(MqttConfigDO config, String clientId, boolean isSub) {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(config.getUsername());
        options.setPassword(config.getPassword().toCharArray());
        options.setServerURIs(new String[]{config.getUrl()});
        options.setAutomaticReconnect(reconnectProperties.isEnabled());
        options.setConnectionTimeout(reconnectProperties.getConnectionTimeout());
        options.setKeepAliveInterval(reconnectProperties.getKeepAliveInterval());// 设置心跳间隔为30秒
        options.setMaxInflight(reconnectProperties.getMaxInflight());
        
        // 注册客户端ID映射
        mqttConnectionListener.registerClientMapping(clientId, config.getId());

        log.info("注册MQTT客户端映射: {} -> 配置ID: {}", clientId, config.getId());
        
        MqttCallback callback = new org.eclipse.paho.client.mqttv3.MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                log.warn("[MQTT-{}] 连接丢失，clientId={}, configId={}, 原因={}", isSub ? "订阅" : "发布", clientId, config.getId(), cause.getMessage(), cause);
                
                // 记录详细的连接断开信息
                log.error("=== MQTT连接断开详细信息 ===");
                log.error("配置ID: {}", config.getId());
                log.error("配置名称: {}", config.getName());
                log.error("客户端ID: {}", clientId);
                log.error("客户端类型: {}", isSub ? "订阅客户端" : "发布客户端");
                log.error("断开时间: {}", java.time.LocalDateTime.now());
                log.error("断开原因: {}", cause.getMessage());
                log.error("异常堆栈: ", cause);
                log.error("=============================");
                
                mqttConnectionListener.handleConnectionLost(clientId);
            }
            @Override
            public void messageArrived(String topic, org.eclipse.paho.client.mqttv3.MqttMessage message) {
                if (isSub) {
                    log.info("[MQTT-订阅] 收到消息，clientId={}, topic={}, payload={}", clientId, topic, new String(message.getPayload()));
                }
            }
            @Override
            public void deliveryComplete(org.eclipse.paho.client.mqttv3.IMqttDeliveryToken token) {
                log.debug("[MQTT-{}] 消息发送完成，clientId={}", isSub ? "订阅" : "发布", clientId);
            }
        };
        return new LoggingMqttPahoClientFactory(options, clientId, callback);
    }

    // 订阅端适配器
    public MqttPahoMessageDrivenChannelAdapter createInboundAdapter(MqttConfigDO config) {
        MqttPahoClientFactory factory = createClientFactory(config, config.getSubClientId(), true);
        String[] topics = config.getSubTopics().split(",");
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                config.getUrl(),
                config.getSubClientId(),
                factory,
                topics
        );
        adapter.setQos(1);
        adapter.setConverter(new DefaultPahoMessageConverter());
        MessageChannel instanceChannel = createInstanceChannel(config.getId(), config.getSubClientId());
        adapter.setOutputChannel(instanceChannel);
        return adapter;
    }

    // 发布端处理器
    public MqttPahoMessageHandler createOutboundHandler(MqttConfigDO config) {
        MqttPahoClientFactory factory = createClientFactory(config, config.getPubClientId(), false);
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(
                config.getUrl(),
                config.getPubClientId(),
                factory
        );
        handler.setDefaultQos(0);
        handler.setDefaultTopic("default");
        handler.setAsync(true);
        
        // 设置消息转换器 - 这是关键修复
        handler.setConverter(new DefaultPahoMessageConverter());
        
        // 注意：某些方法可能在不同版本的Spring Integration中不存在
        // 如果遇到方法不存在的问题，请注释掉相应的行
        
        log.info("创建MQTT出站处理器成功 - 配置ID: {}, 发布客户端ID: {}", config.getId(), config.getPubClientId());
        return handler;
    }

    /**
     * 创建出站消息通道
     */
    public MessageChannel createOutboundChannel(Long configId) {
        return new org.springframework.integration.channel.DirectChannel();
    }

    /**
     * 为每个MQTT实例创建独立的消息通道
     */
    private MessageChannel createInstanceChannel(Long configId, String clientId) {
        DirectChannel channel = new DirectChannel();
        // 创建实例特定的消息处理器
        InstanceSpecificMessageHandler handler = new InstanceSpecificMessageHandler(configId, clientId, mqttMessageService, subscribeWaterService);
        channel.subscribe(handler);
        // 缓存通道和处理器
        inboundChannels.put(configId, channel);
        messageHandlers.put(configId, handler);
        log.info("为MQTT实例创建独立消息通道 - 配置ID: {}, 客户端ID: {}", configId, clientId);
        return channel;
    }

    /**
     * 启动MQTT实例
     */
    public void startMqttInstance(MqttConfigDO config) {
        if (config == null || config.getId() == null) {
            log.error("MQTT配置无效，无法启动实例");
            return;
        }
        
        Long configId = config.getId();
        
        // 如果实例已经运行，先停止
        if (isInstanceRunning(configId)) {
            log.info("MQTT实例已运行，先停止现有实例，配置ID: {}", configId);
            stopMqttInstance(configId);
        }
        
        try {
            // 创建入站适配器（这会创建订阅端工厂）
            MqttPahoMessageDrivenChannelAdapter inboundAdapter = createInboundAdapter(config);
            inboundAdapters.put(configId, inboundAdapter);
            log.info("MQTT入站适配器创建成功 - 配置ID: {}", configId);
            
            // 创建出站处理器（这会创建发布端工厂）
            MqttPahoMessageHandler outboundHandler = createOutboundHandler(config);
            outboundHandlers.put(configId, outboundHandler);
            log.info("MQTT出站处理器创建并存储成功 - 配置ID: {}, 处理器: {}", configId, outboundHandler.getClass().getSimpleName());
            
            // 创建出站消息通道
            MessageChannel outboundChannel = createOutboundChannel(configId);
            outboundChannels.put(configId, outboundChannel);
            
            // 消息通道和处理器已经在createInstanceChannel中创建了
            
            // 启动入站适配器
            try {
                inboundAdapter.start();
                log.info("MQTT实例启动成功，配置ID: {}, 名称: {}", configId, config.getName());
            } catch (Exception e) {
                log.error("启动入站适配器失败，配置ID: {}", configId, e);
                // 清理已创建的资源
                cleanupInstance(configId);
                throw e;
            }
        } catch (Exception e) {
            log.error("MQTT实例启动失败，配置ID: {}, 名称: {}", configId, config.getName(), e);
            // 确保清理资源
            cleanupInstance(configId);
            throw e;
        }
    }

    /**
     * 清理实例资源
     */
    private void cleanupInstance(Long configId) {
        try {
            // 获取配置信息以清理客户端映射
            MqttConfigDO config = getConfigById(configId);
            if (config != null) {
                mqttConnectionListener.unregisterClientMapping(config.getSubClientId());
                mqttConnectionListener.unregisterClientMapping(config.getPubClientId());
            }
            
            clientFactories.remove(configId);
            inboundAdapters.remove(configId);
            outboundHandlers.remove(configId);
            inboundChannels.remove(configId);
            outboundChannels.remove(configId);
            messageHandlers.remove(configId);
        } catch (Exception e) {
            log.warn("清理实例资源时发生异常，配置ID: {}", configId, e);
        }
    }

    /**
     * 停止MQTT实例
     */
    public void stopMqttInstance(Long configId) {
        try {
            // 停止入站适配器
            MqttPahoMessageDrivenChannelAdapter inboundAdapter = inboundAdapters.get(configId);
            if (inboundAdapter != null) {
                try {
                    inboundAdapter.stop();
                } catch (Exception e) {
                    log.warn("停止入站适配器时发生异常，配置ID: {}", configId, e);
                }
                inboundAdapters.remove(configId);
            }
            
            // 清理缓存
            clientFactories.remove(configId);
            outboundHandlers.remove(configId);
            inboundChannels.remove(configId);
            outboundChannels.remove(configId);
            messageHandlers.remove(configId);
            
            log.info("MQTT实例停止成功，配置ID: {}", configId);
        } catch (Exception e) {
            log.error("MQTT实例停止失败，配置ID: {}", configId, e);
        }
    }

    /**
     * 停止所有MQTT实例
     */
    public void stopAllInstances() {
        log.info("开始停止所有MQTT实例...");
        for (Long configId : inboundAdapters.keySet()) {
            try {
                stopMqttInstance(configId);
            } catch (Exception e) {
                log.error("停止MQTT实例失败，配置ID: {}", configId, e);
            }
        }
        log.info("所有MQTT实例停止完成");
    }

    /**
     * 发送消息
     */
    public void sendMessage(Long configId, String topic, String payload) {
        log.info("开始发送MQTT消息 - 配置ID: {}, 主题: {}, 内容: {}", configId, topic, payload);
        
        MqttPahoMessageHandler handler = outboundHandlers.get(configId);
        if (handler == null) {
            log.error("MQTT出站处理器不存在，配置ID: {}", configId);
            log.error("可用的出站处理器配置ID: {}", outboundHandlers.keySet());
            return;
        }
        
        try {
            // 检查处理器状态
            if (!handler.isRunning()) {
                log.warn("MQTT出站处理器未运行，尝试启动，配置ID: {}", configId);
                handler.start();
            }
            
            // 构建消息
            org.springframework.messaging.Message<String> message = org.springframework.integration.support.MessageBuilder
                    .withPayload(payload)
                    .setHeader(org.springframework.integration.mqtt.support.MqttHeaders.TOPIC, topic)
                    .build();
            
            log.debug("发送MQTT消息 - 配置ID: {}, 主题: {}, 内容长度: {}", configId, topic, payload.length());
            
            // 发送消息
            handler.handleMessage(message);
            
            log.info("MQTT消息发送成功 - 配置ID: {}, 主题: {}, 内容: {}", configId, topic, payload);
            
            // 保存发送的消息记录
            mqttMessageService.saveMessage(configId, topic, payload, 0, false, "OUT", null, null);
            
        } catch (Exception e) {
            log.error("MQTT消息发送失败 - 配置ID: {}, 主题: {}, 内容: {}", configId, topic, payload, e);
            
            // 记录详细的错误信息
            log.error("=== MQTT消息发送错误详情 ===");
            log.error("配置ID: {}", configId);
            log.error("主题: {}", topic);
            log.error("内容: {}", payload);
            log.error("错误类型: {}", e.getClass().getSimpleName());
            log.error("错误消息: {}", e.getMessage());
            log.error("处理器状态: {}", handler.isRunning() ? "运行中" : "未运行");
            log.error("处理器类型: {}", handler.getClass().getSimpleName());
            log.error("=============================");
            
            throw new RuntimeException("MQTT消息发送失败", e);
        }
    }

    /**
     * 获取MQTT实例状态
     */
    public boolean isInstanceRunning(Long configId) {
        return inboundAdapters.containsKey(configId) && outboundHandlers.containsKey(configId);
    }

    /**
     * 获取所有运行的实例ID
     */
    public java.util.Set<Long> getRunningInstanceIds() {
        return inboundAdapters.keySet();
    }

    /**
     * 检查MQTT实例连接状态
     */
    public boolean isInstanceConnected(Long configId) {
        MqttPahoMessageDrivenChannelAdapter adapter = inboundAdapters.get(configId);
        if (adapter != null) {
            try {
                return adapter.isRunning() && adapter.getConnectionInfo() != null;
            } catch (Exception e) {
                log.warn("检查MQTT实例连接状态失败，配置ID: {}", configId, e);
                return false;
            }
        }
        return false;
    }

    /**
     * 手动重连MQTT实例
     */
    public void reconnectInstance(Long configId) {
        AtomicBoolean reconnecting = reconnectingFlags.get(configId);
        if (reconnecting != null && reconnecting.get()) {
            log.info("MQTT实例正在重连中，跳过重复重连，配置ID: {}", configId);
            return;
        }

        log.info("开始手动重连MQTT实例，配置ID: {}", configId);
        scheduleReconnect(configId, 0, 1);
    }

    /**
     * 调度重连任务
     */
    private void scheduleReconnect(Long configId, int attempt, int delaySeconds) {
        if (attempt >= reconnectProperties.getMaxAttempts()) {
            log.info("MQTT实例重连失败，已达到最大重连次数，配置ID: {}", configId);
            reconnectingFlags.remove(configId);
            return;
        }

        AtomicBoolean reconnecting = reconnectingFlags.computeIfAbsent(configId, k -> new AtomicBoolean(false));
        if (!reconnecting.compareAndSet(false, true)) {
            log.info("MQTT实例已在重连中，跳过重复调度，配置ID: {}", configId);
            return;
        }

        scheduler.schedule(() -> {
            try {
                log.info("执行MQTT实例重连，配置ID: {}, 第{}次尝试", configId, attempt + 1);
                
                // 停止现有实例
                stopMqttInstance(configId);
                
                // 等待一段时间
                Thread.sleep(2000);
                
                // 重新启动实例
                MqttConfigDO config = getConfigById(configId);
                if (config != null && config.getEnabled()) {
                    startMqttInstance(config);
                    
                    // 检查重连是否成功
                    if (isInstanceConnected(configId)) {
                        log.info("MQTT实例重连成功，配置ID: {}", configId);
                        reconnectingFlags.remove(configId);
                        return;
                    }
                }
                
                // 重连失败，计算下次重连延迟
                int nextDelay = Math.min(delaySeconds * 2, reconnectProperties.getMaxDelay());
                log.warn("MQTT实例重连失败，配置ID: {}, 将在{}秒后重试", configId, nextDelay);
                scheduleReconnect(configId, attempt + 1, nextDelay);
                
            } catch (Exception e) {
                log.error("MQTT实例重连过程中发生异常，配置ID: {}", configId, e);
                reconnectingFlags.remove(configId);
            }
        }, delaySeconds, TimeUnit.SECONDS);
    }

    /**
     * 处理连接断开事件
     */
    public void handleConnectionLost(Long configId) {
        log.warn("MQTT实例连接断开，配置ID: {}", configId);
        
        // 记录详细的连接断开处理信息
        log.error("=== MQTT连接断开处理 ===");
        log.error("配置ID: {}", configId);
        log.error("处理时间: {}", java.time.LocalDateTime.now());
        log.error("处理方式: 自动重连");
        
        // 检查是否启用了自动重连
        MqttConfigDO config = getConfigById(configId);
        if (config != null && config.getEnabled() && reconnectProperties.isEnabled()) {
            log.error("重连配置: 已启用");
            log.error("配置名称: {}", config.getName());
            log.error("重连策略: 自动重连");
            log.info("开始自动重连MQTT实例，配置ID: {}", configId);
            scheduleReconnect(configId, 0, reconnectProperties.getInitialDelay());
        } else {
            log.error("重连配置: 未启用或配置已禁用");
            if (config != null) {
                log.error("配置启用状态: {}", config.getEnabled());
            }
            log.error("重连属性启用状态: {}", reconnectProperties.isEnabled());
        }
        log.error("========================");
    }

    /**
     * 获取重连状态
     */
    public boolean isReconnecting(Long configId) {
        AtomicBoolean reconnecting = reconnectingFlags.get(configId);
        return reconnecting != null && reconnecting.get();
    }

    /**
     * 获取配置信息（用于重连）
     */
    private MqttConfigDO getConfigById(Long configId) {
        try {
            return mqttConfigService.getConfigById(configId);
        } catch (Exception e) {
            log.error("获取MQTT配置失败，配置ID: {}", configId, e);
            return null;
        }
    }

    /**
     * 主动发送 PINGREQ 并等待响应，判断是否在线
     */
    public boolean pingClient(Long configId) {
        log.info("开始PING MQTT客户端 - 配置ID: {}", configId);
        
        // 使用与 sendMessage 相同的方式，通过 outboundHandlers 来发送消息
        MqttPahoMessageHandler handler = outboundHandlers.get(configId);
        if (handler == null) {
            log.error("MQTT出站处理器不存在，无法PING，配置ID: {}", configId);
            log.error("可用的出站处理器配置ID: {}", outboundHandlers.keySet());
            return false;
        }
        
        try {
            // 检查处理器状态
            if (!handler.isRunning()) {
                log.warn("MQTT出站处理器未运行，尝试启动，配置ID: {}", configId);
                handler.start();
            }
            
            // 构建PING消息
            String pingPayload = "ping_" + System.currentTimeMillis();
            org.springframework.messaging.Message<String> message = org.springframework.integration.support.MessageBuilder
                    .withPayload(pingPayload)
                    .setHeader(org.springframework.integration.mqtt.support.MqttHeaders.TOPIC, "ping/test")
                    .build();
            
            log.debug("发送MQTT PING消息 - 配置ID: {}, 主题: ping/test, 内容: {}", configId, pingPayload);
            
            // 发送PING消息
            handler.handleMessage(message);
            
            log.info("MQTT客户端 PING 消息发送成功，配置ID: {}", configId);
            
            // 可选：保存PING消息记录
            try {
                mqttMessageService.saveMessage(configId, "ping/test", pingPayload, 0, false, "PING", null, null);
            } catch (Exception e) {
                log.warn("保存PING消息记录失败，但不影响PING功能，配置ID: {}", configId, e);
            }
            
            return true;
            
        } catch (Exception e) {
            log.warn("MQTT客户端 PING 消息发送失败，配置ID: {}, 错误: {}", configId, e.getMessage());
            
            // 记录详细的错误信息
            log.error("=== MQTT PING 错误详情 ===");
            log.error("配置ID: {}", configId);
            log.error("错误类型: {}", e.getClass().getSimpleName());
            log.error("错误消息: {}", e.getMessage());
            log.error("处理器状态: {}", handler.isRunning() ? "运行中" : "未运行");
            log.error("处理器类型: {}", handler.getClass().getSimpleName());
            log.error("=========================");
            
            return false;
        }
    }

    /**
     * 验证出站处理器状态
     */
    public boolean validateOutboundHandler(Long configId) {
        MqttPahoMessageHandler handler = outboundHandlers.get(configId);
        if (handler == null) {
            log.error("出站处理器不存在，配置ID: {}", configId);
            return false;
        }
        
        try {
            // 检查处理器是否运行
            boolean isRunning = handler.isRunning();
            log.info("出站处理器状态检查 - 配置ID: {}, 运行状态: {}", configId, isRunning);
            
            // 检查处理器类型和基本属性
//            log.info("出站处理器信息 - 配置ID: {}, 类型: {}, 默认QoS: {}, 默认主题: {}",
//                    configId,
//                    handler.getClass().getSimpleName(),
//                    handler.getDefaultQos(),
//                    handler.getDefaultTopic());
            
            return isRunning;
        } catch (Exception e) {
            log.error("验证出站处理器状态失败，配置ID: {}", configId, e);
            return false;
        }
    }
} 